package com.xxl.job.executor.util;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.executor.dto.ApiInfoDto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.utils.URIBuilder;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.net.URI;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author dingfeng
 * @version 1.0
 * @DEC 任务处理工具类
 * @date 2022/3/3 15:03
 */
@Component
@Slf4j
public class TaskUtilV2 {
    @Value("${api.info.mode:public}")
    private String apInfoMode;

    @Value("${task.main.thread:5}")
    private Integer taskMainThread;

    @Value("${task.son.thread:20}")
    private Integer taskSonThread;

    @Autowired
    private JdbcTemplate jdbcTemplate;

    //主任务处理线程池
    private ExecutorService executorService = null;

    @PostConstruct
    public void init() {
        //初始化主任务线程池
        executorService = Executors.newFixedThreadPool(taskMainThread);
    }

    /**
     * 主任务处理方法
     *
     * @param params 调度中心传过来的参数 包含模式名 + 表名
     * @return
     */
    public void handler(String params) throws Exception {
        String[] tableNames = params.split(",");
        //主任务结果集合
         List<Future<TaskReturn>> allTaskReturn = new ArrayList();
        //失败主任务结果集合
         List<TaskReturn> allFailTaskReturn = new ArrayList();
        try {
            for (int i = 0; i < tableNames.length; i++) {
                Future<TaskReturn> submit = executorService.submit(new Task(tableNames[i].trim()));
                allTaskReturn.add(submit);
            }
            while (allTaskReturn.size() != 0) {
                Future<TaskReturn> taskReturnFuture = allTaskReturn.remove(0);
                TaskReturn taskReturn = null;
                taskReturn = taskReturnFuture.get();
                if (taskReturn.isSuccess) {
                    log.info("表{}插入成功，总计数据{}条，耗时{}秒。", taskReturn.getTableName(), taskReturn.getRows(), taskReturn.getTime());
                } else {
                    log.info("表{}插入失败，失败原因{}", taskReturn.getTableName(), taskReturn.getException());
                    allFailTaskReturn.add(taskReturn);
                }
            }

            if (allFailTaskReturn.size() > 0) {
                throw new RuntimeException("存在失败的主任务");
            }
        } catch (Exception e) {
            StringBuffer failTaskTableNames = new StringBuffer("执行失败的表:\n");
            while (allFailTaskReturn.size() != 0) {
                TaskReturn remove = allFailTaskReturn.remove(0);
                failTaskTableNames.append(remove.getTableName() + "\n");
            }
            //让调度中心获取异常信息
            XxlJobHelper.handleFail(failTaskTableNames.toString());
            //可以在这里做补偿机制
        }
    }

    /**
     * 分发批量插入请求
     *
     * @param apiInfoDto
     * @throws Exception
     */
    public TaskReturn distributeHttpGet(ApiInfoDto apiInfoDto) throws Exception {
        /**
         * 先发送一次请求
         * 1.可以判断接口是否通
         * 2.获取到表的数据总量 分批插入
         */
        HashMap<String, String> requestParams = new HashMap<>();
        requestParams.put("pageNumber", "1");
        requestParams.put("pageSize", "1");
        String res = httpGet(apiInfoDto.getApiPath(), requestParams);

        JSONObject parseObject = null;
        JSONObject o = null;
        try {
            parseObject = JSONObject.parseObject(res);
            Map<String, Object> innerMap = parseObject.getInnerMap();
            o = (JSONObject) innerMap.get("data");
        } catch (Exception e) {
            return new TaskReturn()
                    .setException(e)
                    .setIsSuccess(false)
                    .setTableName(apiInfoDto.getTableName());
        }
        //数据总数
        Integer totalNum = (Integer) o.get("totalNum");
        //阿里中台有限制，一次最多只返回2000条数据
        Integer distributeNumber = totalNum % 2000 == 0 ? totalNum / 2000 : totalNum / 2000 + 1;
        AtomicBoolean rollbackFlag = new AtomicBoolean(false);
        AtomicInteger atomicInteger = new AtomicInteger(0);
        ExecutorService distributeExecutorService = Executors.newFixedThreadPool(taskSonThread);
        //主线程 栅栏
        CountDownLatch mainThreadLatch = new CountDownLatch(1);
        long startTime = System.currentTimeMillis();
        for (Integer i = 0; i < distributeNumber; i++) {
            if (rollbackFlag.get()) return new TaskReturn().setIsSuccess(false);
            ApiInfoDto tempApiInfoDto = new ApiInfoDto();
            BeanUtils.copyProperties(apiInfoDto, tempApiInfoDto);
            tempApiInfoDto.setPageNumber(i + 1);
            distributeExecutorService.submit(new DistributeTask(mainThreadLatch, rollbackFlag, tempApiInfoDto, distributeNumber, atomicInteger));
        }
        mainThreadLatch.await();
        distributeExecutorService.shutdown();
        long endTime = System.currentTimeMillis();
        TaskReturn taskReturn = new TaskReturn()
                .setTableName(apiInfoDto.getTableName())
                .setIsSuccess(!rollbackFlag.get())
                .setRows(totalNum).setTime((endTime - startTime) / 1000);

        return taskReturn;
    }

    /**
     * 处理htppGet请求
     */
    public static String httpGet(String apiPath, Map<String, String> params) throws Exception {
        CloseableHttpClient httpClient = HttpPool.getHttpClient();
        URIBuilder uriBuilder = new URIBuilder(apiPath);
        // 定义请求的参数
        if (null != params) {
            Iterator<String> iterator = params.keySet().iterator();
            while (iterator.hasNext()) {
                String key = iterator.next();
                uriBuilder.setParameter(key, params.get(key));
            }
        }
        URI uri = uriBuilder.build();
        HttpGet request = new HttpGet(uri);
        request.setHeader("accept", "*/*");

        CloseableHttpResponse response = null;
        try {
            // 执行请求
            response = httpClient.execute(request);
            // 判断返回状态是否为200
            if (response.getStatusLine().getStatusCode() == 200) {
                //请求体内容
                String content = EntityUtils.toString(response.getEntity(), "UTF-8");
                //内容写入文件
                return content;
            } else {
                log.error("请求{}失败，状态码:{}", apiPath, response.getStatusLine().getStatusCode());
                return "";
            }
        } catch (Exception ex) {
            log.error("请求" + apiPath + "失败", ex);
            throw ex;
        } finally {
            if (response != null) {
                response.close();
            }
            //相当于关闭浏览器
            //httpClient.close();
        }
    }

    /**
     * 解析res,生成sql
     *
     * @param res        response返回结果 json字符串
     * @param table_name
     * @return sql
     */
    public static String handlerResToGetSql(String res, String table_name) {
        //判断fieldNames是否已经拼接好了
        Boolean fieldNamesState = false;
        String resSql = "insert into " + table_name;
        StringBuilder fieldNames = new StringBuilder("(");
        StringBuilder fieldValues = new StringBuilder();
        JSONObject parseObject = JSONObject.parseObject(res);
        Map<String, Object> innerMap = parseObject.getInnerMap();
        JSONObject o = (JSONObject) innerMap.get("data");
        JSONArray rows = (JSONArray) o.get("rows");
        for (int i = 0; i < rows.size(); i++) {
            StringBuffer tempValue = new StringBuffer("(");
            JSONObject obj = rows.getJSONObject(i);
            Set<String> keys = obj.keySet();
            Iterator<String> iterator = keys.iterator();
            while (iterator.hasNext()) {
                String fieldName = iterator.next();
                if (!fieldNamesState) {
                    fieldNames.append("\"" + fieldName + "\",");
                }
                Object fieldValue = obj.get(fieldName);
                if (fieldValue == null || fieldValue instanceof Number) {
                    if (fieldValue instanceof BigDecimal) {
                        fieldValue = ((BigDecimal) fieldValue).toPlainString();
                    }
                    tempValue.append(fieldValue + ",");
                } else {
                    tempValue.append("'" + fieldValue + "',");
                }
            }
            if (!fieldNamesState) {
                fieldNamesState = true;
            }
            fieldValues.append(tempValue.substring(0, tempValue.length() - 1) + "),");
        }
        String fieldNamesRes = fieldNames.substring(0, fieldNames.length() - 1) + ")";
        String fieldValuesRes = fieldValues.substring(0, fieldValues.length() - 1) + ";";

        return resSql + fieldNamesRes + " values" + fieldValuesRes;
    }

    /**
     * 主任务类
     */
    class Task implements Callable<TaskReturn> {
        private String tableName;

        public Task(String tableName) {
            this.tableName = tableName;
        }

        @Override
        public TaskReturn call() throws Exception {
            //1.解析表名称
            if (tableName.indexOf(".") == -1) {
                log.error("表名{}错误,请检查是否有带上模式名", tableName);
                return new TaskReturn()
                        .setIsSuccess(false)
                        .setException(new RuntimeException("表名" + tableName + "错误,请检查是否有带上模式名"))
                        .setTableName(tableName);
            }
            //2.获取表的api相关信息
            String getApiInfoSql = "select * from " + apInfoMode + ".api_info where table_name = ?";
            String[] args = new String[]{tableName};
            Map<String, Object> apiInfoMap = null;
            try {
                apiInfoMap = jdbcTemplate.queryForMap(getApiInfoSql, args);
            } catch (DataAccessException e) {
                return new TaskReturn()
                        .setIsSuccess(false)
                        .setException(e)
                        .setTableName(tableName);
            }
            //3. 执行前
            Object beforeSql = apiInfoMap.get("before_sql");
            if (!Objects.isNull(beforeSql) && ((String) beforeSql).length() > 0) {
                try {
                    jdbcTemplate.update((String) beforeSql);
                    log.info("执行前语句:" + beforeSql + "执行成功");
                } catch (DataAccessException e) {
                    return new TaskReturn()
                            .setIsSuccess(false)
                            .setException(e)
                            .setTableName(tableName);
                }
            }
            //4. 执行
            ApiInfoDto apiInfoDto = new ApiInfoDto()
                    .setTableName((String) apiInfoMap.get("table_name"))
                    .setApiPath((String) apiInfoMap.get("api_path"))
                    .setPageSize(2000);
            TaskReturn taskReturn = distributeHttpGet(apiInfoDto);
            if (!taskReturn.isSuccess) {
                log.error("子任务执行出错,清空表数据");
                try {
                    jdbcTemplate.update("TRUNCATE TABLE " + apiInfoDto.getTableName());
                    // 清空表后,终止程序运行
                    return taskReturn;
                } catch (DataAccessException e) {
                    return new TaskReturn()
                            .setIsSuccess(false)
                            .setException(e)
                            .setTableName(tableName);
                }
            }
            //5. 执行后
            Object afterSql = apiInfoMap.get("after_sql");
            if (!Objects.isNull(afterSql) && ((String) afterSql).length() > 0) {
                try {
                    jdbcTemplate.update((String) afterSql);
                    log.info("执行后语句:" + afterSql + "执行成功");
                } catch (DataAccessException e) {
                    return new TaskReturn()
                            .setIsSuccess(false)
                            .setException(e)
                            .setTableName(tableName);
                }
            }
            return taskReturn;
        }
    }

    /**
     * 子任务类
     */
    class DistributeTask implements Callable<Boolean> {

        private AtomicBoolean rollbackFlag;
        private ApiInfoDto apiInfoDto;
        private CountDownLatch mainThreadLatch;
        private Integer distributeNumber;
        private AtomicInteger successNumber;

        public DistributeTask(CountDownLatch mainThreadLatch, AtomicBoolean rollbackFlag, ApiInfoDto apiInfoDto, Integer distributeNumber, AtomicInteger successNumber) {
            this.mainThreadLatch = mainThreadLatch;
            this.rollbackFlag = rollbackFlag;
            this.apiInfoDto = apiInfoDto;
            this.distributeNumber = distributeNumber;
            this.successNumber = successNumber;
        }

        @Override
        public Boolean call() throws Exception {
            //封装请求参数
            HashMap<String, String> requestParams = new HashMap<>();
            requestParams.put("pageNumber", apiInfoDto.getPageNumber() + "");
            requestParams.put("pageSize", apiInfoDto.getPageSize() + "");
            /**
             * 判断回滚标志
             * 1.如果为true,则不再请求
             *  但是在高并发情况下 不能控制代码的原子性
             *  需要在finally里再次判断,如果回滚标志被设置为true
             *  则需要执行相应的操作删除脏数据
             */
            if (rollbackFlag.get()) {
                mainThreadLatch.countDown();
                return false;
            }
            //发送请求
            String res = httpGet(apiInfoDto.getApiPath(), requestParams);

            //解析出sql
            String sql = handlerResToGetSql(res, apiInfoDto.getTableName());
            long start = System.currentTimeMillis(), end = 0, update = 0;
            //执行sql
            try {
                update = jdbcTemplate.update(sql);
                end = System.currentTimeMillis();
            } catch (DataAccessException e) {
                //执行失败,设置回滚标志
                rollbackFlag.set(true);
                mainThreadLatch.countDown();
                log.error("插入第" + requestParams.get("pageNumber") + "页数据失败", e);
                return false;
            } finally {
                //删除脏数据
                if (rollbackFlag.get()) {
                    jdbcTemplate.update("truncate table " + apiInfoDto.getTableName());
                }
                int i = successNumber.addAndGet(1);
                if (distributeNumber == i) {
                    mainThreadLatch.countDown();
                }
            }
            return true;
        }
    }

    /**
     * 任务返回类的包装对象
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    @Accessors(chain = true)
    class TaskReturn {
        //表名称
        private String tableName;
        //影响行数
        private Integer rows;
        //执行时间
        private Long time;
        //执行是否成功
        private Boolean isSuccess;
        //失败抛出的异常
        private Exception exception;
    }
}



